package cn.fan.vertd.verticle;

import cn.fan.vertd.utils.TunerKafka;
import io.vertx.core.AbstractVerticle;
import io.vertx.core.Future;
import io.vertx.core.Promise;
import io.vertx.core.json.JsonObject;
import io.vertx.kafka.client.producer.KafkaProducerRecord;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.util.Random;

public class KafkaProducerVerticle extends AbstractVerticle {

    private static final Logger log = LogManager.getLogger();

    final TunerKafka kafka = new TunerKafka();

    @Override
    public void start(Promise<Void> startPromise) throws Exception {
        init(config())
                .compose(this::pushMessage)
                .onSuccess(startPromise::complete)
                .onFailure(startPromise::fail);

    }

    @Override
    public void stop(Promise<Void> stopPromise) throws Exception {
        super.stop(stopPromise);
    }

    private Future<Void> init(JsonObject conf) {
        Promise<Void> promise = Promise.promise();
        kafka.createProducer(vertx, conf);
        promise.complete();
        return promise.future();
    }

    private Future<Void> pushMessage(Void unused) {
        Promise<Void> promise = Promise.promise();
        vertx.setPeriodic(1000, timerId->{
            Random random = new Random();
            String content = "test" + random.nextDouble();
            log.info("push content: {}", content);
            KafkaProducerRecord<String, String> record = KafkaProducerRecord.create(TunerKafka.TEST_TOPIC, content);
            kafka.getProducer().send(record);
        });
        promise.complete();
        return promise.future();
    }
}
